package com.atguigu.day03;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Timestamp;

// 定时器举例
public class Example8 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .socketTextStream("localhost", 9999)
                .keyBy(r -> r)
                .process(new KeyedProcessFunction<String, String, String>() {
                    @Override
                    public void processElement(String in, Context ctx, Collector<String> out) throws Exception {
                        // 获取当前机器时间
                        long currTs = ctx.timerService().currentProcessingTime();
                        // 30s之后的时间戳
                        long thirtySecondsLater = currTs + 30 * 1000L;
                        // 60s之后的时间戳
                        long sixtySecondsLater = currTs + 60 * 1000L;
                        // 注册定时器
                        // 注册的是输入数据in的key所对应的定时器
                        ctx.timerService().registerProcessingTimeTimer(thirtySecondsLater);
                        ctx.timerService().registerProcessingTimeTimer(sixtySecondsLater);

                        out.collect("key: " + ctx.getCurrentKey() + ", 数据：" + in + "," +
                                "到达的时间是：" + new Timestamp(currTs) + "," +
                                "注册的第一个定时器的时间戳：" + new Timestamp(thirtySecondsLater) + "," +
                                "注册的第二个定时器的时间戳：" + new Timestamp(sixtySecondsLater));
                    }

                    @Override
                    public void onTimer(long timerTs, OnTimerContext ctx, Collector<String> out) throws Exception {
                        out.collect("key: " + ctx.getCurrentKey() + "的定时器触发了，" +
                                "定时器的时间戳是：" + new Timestamp(timerTs) + "," +
                                "定时器真正的执行时间是：" + new Timestamp(ctx.timerService().currentProcessingTime()));
                    }
                })
                .print();


        env.execute();
    }
}
